package com.dongnao.network.nio.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

/**
 * 单线程版的Reactor模式
 * Reactor模式，就是将原来一个人干的事情进行分类拆分，拆分成适合多个岗位的人干。
 * 比如连接请求处理，交给Acceptor；客户端数据读取写入处理，交给Handler。
 */
public class SingleReactor {

    public static void main(String[] args) {
        try {
            Reactor reactor = new Reactor(8080);
            reactor.run();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 处理多个客户端并发请求
     */
    private static class Reactor implements Runnable {
        final Selector sel;
        final ServerSocketChannel serverSocket;

        public Reactor(int port) throws IOException {
            sel = Selector.open();
            serverSocket = ServerSocketChannel.open();
            serverSocket.configureBlocking(false);
            //获取serverSocket在Selector中注册的标识sk，sk绑定了serverSocket与Selector的关系，维护channel事件
            SelectionKey sk = serverSocket.register(sel, SelectionKey.OP_ACCEPT);
            serverSocket.socket().bind(new InetSocketAddress(port));
//            serverSocket.bind(new InetSocketAddress(port));
            //附加一个Runnable的Accept事件处理者
            sk.attach(new Acceptor(sel, serverSocket));
        }

        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    sel.select();
                    Set<SelectionKey> selKeySet = sel.selectedKeys();
                    Iterator<SelectionKey> it = selKeySet.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        System.out.println("key isAcceptable: " + key.isAcceptable()
                                + ", isConnectable: " + key.isConnectable()
                                + ", isReadable: " + key.isReadable()
                                + ", isWritable: " + key.isWritable()
                                + ", isValid: " + key.isValid());
                        dispatch(key);
                    }
                    selKeySet.clear();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        private void dispatch(SelectionKey selectionKey) {
            Runnable r = (Runnable) selectionKey.attachment();
            if (r != null) {
                r.run();
            }
        }
    }

    private static class Acceptor implements Runnable {
        final Selector sel;
        final ServerSocketChannel serverSocket;

        public Acceptor(Selector selector, ServerSocketChannel serverSocketChannel) {
            sel = selector;
            serverSocket = serverSocketChannel;
        }

        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocket.accept();
                System.out.println("----Acceptor#run()----channel: " + channel);
                if (channel != null) {
                    System.out.println("收到新的客户端连接：" + channel);
                    new Handler(sel, channel);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 客户端读写事件处理器
     */
    private static class Handler implements Runnable {
        final SocketChannel socket;
        final SelectionKey sk;
        final int MAXIN = 1024;
        final int MAXOUT = 1024;
        ByteBuffer input = ByteBuffer.allocate(MAXIN);
        ByteBuffer output = ByteBuffer.allocate(MAXOUT);
        static final int READING = 0, SENDINGINT = 1;
        int state = READING;

        public Handler(Selector selector, SocketChannel socketChannel) throws IOException {
            System.out.println("----Handler()----");
            socket = socketChannel;
            socket.configureBlocking(false);
            //获取客户端通道与Selector的绑定key
            sk = socket.register(selector, 0);
            sk.interestOps(SelectionKey.OP_READ);//读事件感兴趣
            sk.attach(this);//绑定自己为读写事件的处理器
            selector.wakeup();//激活在select()方法上的线程，告诉它感兴趣的事件发生变化了
        }

        @Override
        public void run() {
            System.out.println("----Handler#run()----");
            try {
                if (state == READING) {
                    read();
                } else if (state == SENDINGINT) {
                    send();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        void read() throws IOException {
            socket.read(input);
            if (inputIsComplete()) {
                input.flip();
                process();//读完数据，进行业务逻辑处理
                input.clear();
                state = SENDINGINT;
                //注册感兴趣写事件
                sk.interestOps(SelectionKey.OP_WRITE);
            }
        }

        void send() throws IOException {
            output.flip();
            socket.write(output);
            if (outputIsComplete()) {
                output.clear();
                //只有写事件操作完成了，才取消
                sk.cancel();
            }
        }

        private boolean inputIsComplete() {
            return input.position() > 0;
        }

        private boolean outputIsComplete() {
            return !output.hasRemaining();
        }

        void process() {
            //模拟业务逻辑处理
            System.out.println("收到数据：" + new String(input.array(), input.position(), input.limit()));
            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(300));
            String msg = "Now time is : " + (new Date()).toString();
            output.put(msg.getBytes());
        }
    }
}
